#pragma once

#include <leveldb/db.h>

#include <atomic>
#include <cassert>
#include <iostream>
#include <mutex>

#include "KeeperMessage.h"
#include "NuRaft.h"

namespace keeper_server {

class KeeperStateMachine : public nuraft::state_machine {
 public:
  KeeperStateMachine(const std::string& path) {
    last_committed_idx_ = 0;
    leveldb_path = path;
    ldbptr = nullptr;
  }

  ~KeeperStateMachine() {}

  NuRaftBufferPtr pre_commit(const ulong log_idx, NuRaftBuffer& data) {
    // Extract string from `data.
    NuRaftBufferSerializer bs(data);
    std::string str = bs.get_str();

    // Just print.
    std::cout << "pre_commit " << log_idx << ": " << str << std::endl;
    return nullptr;
  }

  NuRaftBufferPtr commit(const ulong log_idx, NuRaftBuffer& data) {
    // Extract string from `data.
    NuRaftBufferSerializer bs(data);
    auto msg_type = KeeperMessageType(bs.get_i32());
    switch (msg_type) {
      case KEEPER_MESSAGE_TYPE_START: {
        std::cout << "Get KEEPER_MESSAGE_TYPE_START message" << std::endl;
      } break;
      case KV_MESSAGE: {
        auto kv_msg = std::make_shared<KVMessage>();
        kv_msg->deserlialize_message(data);
        kv_msg->printInfo();
        write_leveldb(kv_msg->get_key(), kv_msg->get_value());
      } break;
      case STR_MESSAGE: {
        auto str_msg = std::make_shared<StrMessage>();
        str_msg->deserlialize_message(data);
        str_msg->printInfo();
      } break;
      default: {
        std::cout << "Get unknown message" << msg_type << std::endl;
      }
    }

    // Update last committed index number.
    last_committed_idx_ = log_idx;
    return nullptr;
  }

  void commit_config(const ulong log_idx, NuRaftClusterConfigPtr& new_conf) {
    // Nothing to do with configuration change. Just update committed index.
    last_committed_idx_ = log_idx;
  }

  void rollback(const ulong log_idx, NuRaftBuffer& data) {
    // Extract string from `data.
    NuRaftBufferSerializer bs(data);
    std::string str = bs.get_str();

    // Just print.
    std::cout << "rollback " << log_idx << ": " << str << std::endl;
  }

  int read_logical_snp_obj(NuRaftSnapshot& s, void*& user_snp_ctx, ulong obj_id,
                           NuRaftBufferPtr& data_out, bool& is_last_obj) {
    // Put dummy data.
    data_out = NuRaftBuffer::alloc(sizeof(nuraft::int32));
    NuRaftBufferSerializer bs(data_out);
    bs.put_i32(0);

    is_last_obj = true;
    return 0;
  }

  void save_logical_snp_obj(NuRaftSnapshot& s, ulong& obj_id,
                            NuRaftBuffer& data, bool is_first_obj,
                            bool is_last_obj) {
    std::cout << "save snapshot " << s.get_last_log_idx() << " term "
              << s.get_last_log_term() << " object ID " << obj_id << std::endl;
    // Request next object.
    obj_id++;
  }

  bool apply_snapshot(NuRaftSnapshot& s) {
    std::cout << "apply snapshot " << s.get_last_log_idx() << " term "
              << s.get_last_log_term() << std::endl;
    // Clone snapshot from `s`.
    {
      std::lock_guard<std::mutex> l(last_snapshot_lock_);
      NuRaftBufferPtr snp_buf = s.serialize();
      last_snapshot_ = NuRaftSnapshot::deserialize(*snp_buf);
    }
    return true;
  }

  void free_user_snp_ctx(void*& user_snp_ctx) {}

  NuRaftSnapshotPtr last_snapshot() {
    // Just return the latest snapshot.
    std::lock_guard<std::mutex> l(last_snapshot_lock_);
    return last_snapshot_;
  }

  ulong last_commit_index() { return last_committed_idx_; }

  void create_snapshot(NuRaftSnapshot& s,
                       nuraft::async_result<bool>::handler_type& when_done) {
    std::cout << "create snapshot " << s.get_last_log_idx() << " term "
              << s.get_last_log_term() << std::endl;
    // Clone snapshot from `s`.
    {
      std::lock_guard<std::mutex> l(last_snapshot_lock_);
      NuRaftBufferPtr snp_buf = s.serialize();
      last_snapshot_ = NuRaftSnapshot::deserialize(*snp_buf);
    }
    nuraft::ptr<std::exception> except(nullptr);
    bool ret = true;
    when_done(ret, except);
  }

  void init() {
    if (!leveldb_path.empty()) {
      leveldb::Options options;
      options.create_if_missing = true;
      leveldb::Status status =
          leveldb::DB::Open(options, leveldb_path, &ldbptr);
      if (!status.ok()) {
        std::cerr << "open level db error:" << status.ToString() << std::endl;
      }
    }
  }

  void write_leveldb(const std::string& k, const std::string& v) {
    leveldb::WriteOptions putOptions;
    putOptions.sync = true;
    // const leveldb::Slice s_key = leveldb::Slice(key);
    // const leveldb::Slice s_value = leveldb::Slice(key);
    auto status = ldbptr->Put(putOptions, k, v);
    if (!status.ok()) {
      std::cerr << "put data error:" << status.ToString() << std::endl;
    }
    std::cout << "put data successfully!" << std::endl;
  }

  std::string read_leveldb(const std::string& k) {
    leveldb::ReadOptions getOptions;
    std::string v;
    auto status = ldbptr->Get(getOptions, k, &v);
    if (!status.ok()) {
      std::cerr << "get data error:" << status.ToString() << std::endl;
    }
    std::cout << "get data successfully:" << v << std::endl;

    return std::move(v);
  }

 private:
  // Last committed Raft log number.
  std::atomic<uint64_t> last_committed_idx_;

  // Last snapshot.
  NuRaftSnapshotPtr last_snapshot_;

  // Mutex for last snapshot.
  std::mutex last_snapshot_lock_;
  std::string leveldb_path;
  leveldb::DB* ldbptr;
};
};  // namespace keeper_server
